# remoteaccount.pyi
import logging
from typing import Any, Callable, ContextManager, Iterator, Optional

class RemoteAccountSSHConfig:
    host: Optional[str]
    hostname: str
    user: Optional[str]
    port: int
    password: Optional[str]
    identityfile: Optional[str]
    connecttimeout: Optional[int]

    def __init__(
        self,
        host: Optional[str] = ...,
        hostname: Optional[str] = ...,
        user: Optional[str] = ...,
        port: Optional[int] = ...,
        password: Optional[str] = ...,
        identityfile: Optional[str] = ...,
        connecttimeout: Optional[int] = ...,
        **kwargs: Any,
    ) -> None: ...
    @staticmethod
    def from_string(config_str: str) -> "RemoteAccountSSHConfig": ...
    def to_json(self) -> dict: ...
    def __repr__(self) -> str: ...
    def __eq__(self, other: Any) -> bool: ...
    def __hash__(self) -> int: ...

class RemoteAccountError(Exception):
    account_str: str
    msg: str | bytes

    def __init__(self, account: Any, msg: str) -> None: ...
    def __str__(self) -> str: ...

class RemoteCommandError(RemoteAccountError):
    exit_status: int
    cmd: str

    def __init__(self, account: Any, cmd: str, exit_status: int, msg: str) -> None: ...
    def __str__(self) -> str: ...

class RemoteAccount:
    ssh_config: RemoteAccountSSHConfig
    # in principle hostname or ssh_hostname could be None, i.e., the
    # constructor does not exclude that possibility, but in practice
    # these are always non-None for us, see ducktape_cluster.json.j2
    # in this repo (for docker) and vtools (for CDT), so we type them
    # as str here, to avoid constant pointless hoop jumping later on
    # to deal with their possible-None-ness-but-not-really-None-ness.
    hostname: str
    ssh_hostname: str
    user: Optional[str]
    externally_routable_ip: Optional[str]
    os: Any
    _ssh_client: Any
    _sftp_client: Any
    _custom_ssh_exception_checks: list
    _logger: Optional[logging.Logger]

    def __init__(
        self,
        ssh_config: RemoteAccountSSHConfig,
        externally_routable_ip: Optional[str] = ...,
        logger: Optional[logging.Logger] = ...,
        ssh_exception_checks: list = ...,
    ) -> None: ...
    @property
    def operating_system(self) -> Any: ...
    @property
    def logger(self) -> logging.Logger: ...
    @logger.setter
    def logger(self, logger: logging.Logger) -> None: ...
    def _log(self, level: int, msg: str, *args: Any, **kwargs: Any) -> None: ...
    def _set_ssh_client(self) -> None: ...
    @property
    def ssh_client(self) -> Any: ...
    def _set_sftp_client(self) -> None: ...
    @property
    def sftp_client(self) -> Any: ...
    def close(self) -> None: ...
    def __str__(self) -> str: ...
    def __repr__(self) -> str: ...
    def __eq__(self, other: Any) -> bool: ...
    def __hash__(self) -> int: ...
    def wait_for_http_service(
        self, port: int, headers: Any, timeout: int = ..., path: str = ...
    ) -> None: ...
    def _can_ping_url(self, url: str, headers: Any) -> bool: ...
    def available(self) -> bool: ...
    def ssh(self, cmd: str, allow_fail: bool = ...) -> int: ...
    def ssh_capture(
        self,
        cmd: str,
        allow_fail: bool = ...,
        callback: Optional[Callable[[str], Any]] = ...,
        combine_stderr: bool = ...,
        timeout_sec: Optional[float] = ...,
    ) -> "SSHOutputIter": ...
    def ssh_output(
        self,
        cmd: str,
        allow_fail: bool = ...,
        combine_stderr: bool = ...,
        timeout_sec: Optional[float] = ...,
    ) -> bytes: ...
    def alive(self, pid: int) -> bool: ...
    def signal(self, pid: int, sig: int, allow_fail: bool = ...) -> None: ...
    def kill_process(
        self, process_grep_str: str, clean_shutdown: bool = ..., allow_fail: bool = ...
    ) -> None: ...
    def java_pids(self, match: str) -> list[int]: ...
    def kill_java_processes(
        self, match: str, clean_shutdown: bool = ..., allow_fail: bool = ...
    ) -> None: ...
    def copy_between(self, src: str, dest: str, dest_node: Any) -> None: ...
    def scp_from(self, src: str, dest: str, recursive: bool = ...) -> None: ...
    def _re_anchor_basename(self, path: str, directory: str) -> str: ...
    def copy_from(self, src: str, dest: str) -> None: ...
    def scp_to(self, src: str, dest: str, recursive: bool = ...) -> None: ...
    def copy_to(self, src: str, dest: str) -> None: ...
    def islink(self, path: str) -> bool: ...
    def isdir(self, path: str) -> bool: ...
    def exists(self, path: str) -> bool: ...
    def isfile(self, path: str) -> bool: ...
    def open(self, path: str, mode: str = ...) -> Any: ...
    def create_file(self, path: str, contents: str) -> None: ...
    def mkdir(self, path: str, mode: int = ...) -> None: ...
    def mkdirs(self, path: str, mode: int = ...) -> None: ...
    def remove(self, path: str, allow_fail: bool = ...) -> None: ...
    def monitor_log(self, log: str) -> ContextManager["LogMonitor"]: ...

class SSHOutputIter:
    iter_obj_func: Callable[[], Iterator[str]]
    channel_file: Any

    def __init__(
        self, iter_obj_func: Callable[[], Iterator[str]], channel_file: Any = ...
    ) -> None: ...
    def __iter__(self) -> "SSHOutputIter": ...
    def next(self) -> str: ...
    def __next__(self) -> str: ...
    def has_next(self, timeout_sec: Optional[float] = ...) -> bool: ...

class LogMonitor:
    acct: RemoteAccount
    log: str
    offset: int

    def __init__(self, acct: RemoteAccount, log: str, offset: int) -> None: ...
    def wait_until(self, pattern: str, **kwargs: Any) -> Any: ...

class IgnoreMissingHostKeyPolicy:
    def missing_host_key(self, client: Any, hostname: str, key: Any) -> None: ...
